package com.atguigu.chapter07;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.time.Duration;
import java.util.Properties;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/3 14:19
 */
public class Flink17_Watermark_Kafka {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // TODO Source - 从 Kafka 读

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop1:9092,hadoop2:9092,hadoop3:9092");
        properties.setProperty("group.id", "fffffffffff");


        FlinkKafkaConsumer<String> kafkaSourceFunction = new FlinkKafkaConsumer<>(
                "flink0923",
                new SimpleStringSchema(),
                properties
        );

        // TODO 如果source是 Kafka，建议在 官方的SourceFunction 指定watermark生成
        kafkaSourceFunction.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .forBoundedOutOfOrderness(Duration.ofSeconds(2))
                // TODO 这里不需要指定时间戳的提取，会自动提取kafka的时间戳
        );

        // TODO Kafka经典问题： 全局有序
        // Flink使用watermark机制，一定程度上来 处理 kafka 的全局有序性


        env
                .addSource(kafkaSourceFunction)
//                .assignTimestampsAndWatermarks() 不建议
                .print();


        env.execute();
    }
}
/*
    补充扩展： 为什么要关注Kafka的全局有序性？
        如果 是用 canal或maxwell等工具，同步 关系型数据库的 数据到kafka，那么就要关注 操作顺序：
            表 test，某行数据做了如下操作： ① 插入  ② 更新1  ③更新2  ④删除
            => 操作顺序不能乱，否则会出现问题
        Kafka是全局无序、单分区有序，怎么办？
            思路： 把 同一张表的 数据 ，都 发送到 同一个分区
            实现： Kafka的生产者 API，可以指定 key
                  key取hash值，对kafka分区数 取模
                  我们可以指定 key为 库名+表名

        如果 一个 Topic只有一张表的数据，怎么办？
            那么这个Topic可以设为 单分区

     Flink可以依赖 watermark机制，一定程度 处理 乱序问题
 */